package cn.xuexiyuan.flinkstudy.sql;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description:  使用事件时间 + watermark + flink SQL API 中的 window 来实现, 最近5秒每个用户每个渠道订单数,订单最大金额,订单最小基恩,订单金额 统计
 *
 * @Author 左龙龙
 * @Date 21-3-30
 * @Version 1.0
 **/
public class Demo04 {

    public static void main(String[] args) throws Exception{
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            private boolean running = true;
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while(running){
                    Order order = DataFactory.createRandomOrder(3, 5);
                    System.out.println("->" + order);
                    ctx.collect(order);

                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        // 2.transformation
        SingleOutputStreamOperator<Order> orderWatermarkDS = orderDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((order, timestamp) -> order.getOrder_time())
        );
        tableEnv.createTemporaryView("t_order", orderWatermarkDS, $("user_id"), $("channel_keyword"), $("payment_money"), $("order_time").rowtime());

        String sql = "select " +
                        "user_id, channel_keyword, " +
                        "count(1) as order_num, " +
                        "max(payment_money) as max_money," +
                        "min(payment_money) as min_money," +
                        "sum(payment_money) as sum_money " +
                    "from t_order " +
                    "group by user_id, channel_keyword, tumble(order_time, INTERVAL '5' SECOND)";

        Table resultTable = tableEnv.sqlQuery(sql);
        resultTable.printSchema();

        DataStream<Tuple2<Boolean, Row>> resultDS = tableEnv.toRetractStream(resultTable, Row.class);



        // 3.sink
        resultDS.print();

        // 4.excute
        env.execute();

    }
}